package com.demo.kafka.config;

import com.demo.kafka.constants.DelayTopicEnum;
import com.demo.kafka.entity.KafkaMessage;
import com.demo.kafka.processor.KafkaDelayMsgProcessor;
import com.demo.kafka.processor.KafkaTimerTaskProcessor;
import com.demo.kafka.producer.KafkaDelayProducer;
import kafka.utils.timer.SystemTimer;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.context.annotation.Configuration;
import org.springframework.lang.NonNull;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * kafka延时消息配置
 * @author cyj
 **/
@Slf4j
@Configuration
public class KafkaConsumeDelayMsgConfig implements SmartLifecycle, ApplicationContextAware, SmartInitializingSingleton {

    private boolean running;

    private ApplicationContext applicationContext;

    private KafkaDelayProducer delayProducer;
    private KafkaDelayMsgProcessor kafkaDelayMsgProcessor;
    private final ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(10000);
    private Map<String, Consumer<String, KafkaMessage>> consumerMap;

    /**
     * 驱动时间轮运行线程池
     */
    private ExecutorService timeWheelScheduler;
    /**
     * 执行时间轮到期的事件线程池
     */
    private ExecutorService queueConsumerScheduler;

    public KafkaConsumeDelayMsgConfig() {
        log.info("KafkaConsumeDelayMsgConfig 初始化");
    }

    @Override
    public void setApplicationContext(@NonNull ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    /**
     * 获取消费者和定义topic
     * 获取生产者
     * 初始化线程池
     */
    @Override
    public void afterSingletonsInstantiated() {
        consumerMap = applicationContext.getBean("delayConsumerMap", Map.class);

        delayProducer = applicationContext.getBean(KafkaDelayProducer.class);

        timeWheelScheduler = new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                new CustomizableThreadFactory("时间轮线程池"));

        queueConsumerScheduler = new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                new CustomizableThreadFactory("执行时间轮到期事件线程池"));
    }

    /**
     * 开始执行延时消息逻辑
     * 1.获取监听分区并且设定每个分区在时间轮的延迟执行时间
     * 2.初始化定时任务，分别是：
     *  a.获取消费者监听的分区
     *  b.推动时间轮执行
     *  c.执行到期事件
     */
    @Override
    public void start() {
        if (isRunning()) {
            return;
        }

        // 时间轮数据初始化
        initTimeWheelData();

        // 1000ms推动一次时间轮运转
        this.timeWheelScheduler.submit(() -> {
            SystemTimer systemTimer = KafkaTimerTaskProcessor.getSystemTimer();
            for (; ; ) {
                systemTimer.advanceClock(1000);
            }
        });

        this.queueConsumerScheduler.submit(() -> {
            for (; ; ) {
                String topic = queue.take();
                kafkaDelayMsgProcessor.processDelayMsgAndGetPartitionDelayTime(topic);
                kafkaDelayMsgProcessor.addTimeTask(List.of(DelayTopicEnum.getDelayTopicEnum(topic)), queue);
            }
        });

        this.running = true;
    }

    private void initTimeWheelData() {
        kafkaDelayMsgProcessor = new KafkaDelayMsgProcessor(delayProducer, consumerMap);
        kafkaDelayMsgProcessor.addTimeTask(queue);
    }

    @Override
    public void stop() {
        consumerMap.forEach((k,v) -> v.close());
    }

    @Override
    public boolean isRunning() {
        return running;
    }
}